package com.ysy.mq;

import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.google.common.util.concurrent.RateLimiter;
import com.ysy.constant.RedisConstant;
import com.ysy.entity.Fans;
import com.ysy.entity.Follow;
import com.ysy.entity.dto.FollowCancelCountDTO;
import com.ysy.entity.dto.FollowDTO;
import com.ysy.enums.FollowCancelTypeEnum;
import com.ysy.mapper.FansMapper;
import com.ysy.mapper.FollowMapper;
import com.ysy.persimmons.common.util.DateUtils;
import jakarta.annotation.Resource;

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import com.ysy.constant.RocketMqConstant;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionTemplate;

import java.util.Collections;
import java.util.Objects;

/**
 * 顺序消费
 */
@Component
@RocketMQMessageListener(consumerGroup = "persimmons_group",
        topic = RocketMqConstant.TOPIC_FOLLOW_CANCEL,
        messageModel = MessageModel.CLUSTERING,
        consumeMode = ConsumeMode.ORDERLY)
public class FollowAndCancelConsumer implements RocketMQListener<Message> {

    private final Logger logger = LoggerFactory.getLogger(FollowAndCancelConsumer.class);



    @Resource
    private RateLimiter rateLimiter ;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Resource
    private FollowMapper followMapper;
    @Resource
    private FansMapper fansMapper;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private TransactionTemplate transactionTemplate;

    @Override
    public void onMessage(Message message) {

        if (Objects.isNull(message)) {
            logger.warn("FollowAndCancelConsumer--> 接受到的消息 is null ");
            return;
        }

        // 下游稳定
        rateLimiter.acquire();

        // 消息体
        String body = new String(message.getBody());
        String tags = message.getTags();
        logger.info(" FollowAndCancelConsumer ==> 接收到了消息:{}, tags:{}", body, tags);
        FollowDTO followDTO = JSON.parseObject(body, FollowDTO.class);

        switch (tags) {
            case RocketMqConstant.FOLLOW:
                followUserHandle(followDTO);
                break;
            case RocketMqConstant.CANCEL:
                cancelFollowHandle(followDTO);
                break;
            default: logger.info("FollowAndCancelConsume未知的类型：{}",tags);
        }
    }


    /**
     * 关注
     * @param body
     */
    private void followUserHandle(FollowDTO body) {

      boolean result = Boolean.TRUE.equals(transactionTemplate.execute(status ->  {
          try{
              Follow follow = new Follow();
              follow.setUserId(body.getUserId());
              follow.setFollowUserId(body.getUserId());
              follow.setCreateTime(body.getCreateTime());
              followMapper.insert(follow);

              Fans fans = new Fans();
              fans.setUserId(body.getUserId());
              fans.setFansUserId(body.getUserId());
              fans.setCreateTime(body.getCreateTime());
              fansMapper.insert(fans);
              return true;
          }catch (Exception e){
              status.setRollbackOnly();
              logger.error("插入关注和粉丝表失败：",e);
          }
          return false;
      }));


        logger.info("插入关注和粉丝表状态：{}",result);
        if (!result){
            return;
        }
        // 《《---更新redis的粉丝数量，如果过期不存在就会导致redis中没有，过期有数据库兜底---》》
        DefaultRedisScript<Long> script = new DefaultRedisScript<Long>();
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/z_set_follow_check_update_fans.lua")));
        script.setResultType(Long.class);
        String fansRedisKey = RedisConstant.buildUserFansKey(body.getFollowUserId());
        // 时间排序
        long scope = DateUtils.localDateTimeTimestamp(body.getCreateTime());
        Long execute = redisTemplate.execute(script, Collections.singletonList(fansRedisKey), body.getUserId(), scope);
        logger.info("更新redis粉丝数量结果：{}",execute);

        // 同步计数服务
        FollowCancelCountDTO dto = FollowCancelCountDTO.builder()
                .userId(body.getUserId())
                .followUserId(body.getFollowUserId())
                .type(FollowCancelTypeEnum.FOLLOW.getType())
                .build();
        sendCountMq(dto);
    }


    /**
     * 取消
     * @param body
     */
    private void cancelFollowHandle(FollowDTO body) {


       boolean delResult = Boolean.TRUE.equals(transactionTemplate.execute(status -> {
            try{

                Follow follow = new Follow();
                follow.setUserId(body.getUserId());
                follow.setFollowUserId(body.getUserId());
                followMapper.deleteById(follow);

                Fans fans = new Fans();
                fans.setUserId(body.getFollowUserId());
                fans.setFansUserId(body.getUserId());
                fansMapper.deleteById(fans);

                return true;
            }catch (Exception e){
                status.setRollbackOnly();
                logger.error("删除关注和粉丝表失败：",e);
            }
            return false;
        }));

        logger.info("删除关注和粉丝表结果：{}",delResult);
        if (!delResult){
            // 触发报警或者后续重试
            return;
        }
        // 更新redis的粉丝数量
        String fansRedisKey = RedisConstant.buildUserFansKey(body.getFollowUserId());
        // 删除指定粉丝,这里如果缓存过期了会不存在，靠数据库兜底
        // 返回值是删除的数量,为空是0
        Long remove = redisTemplate.opsForZSet().remove(fansRedisKey, body.getUserId());
        logger.info("redis删除粉丝结果：{}",remove);

        // 同步计数服务
        FollowCancelCountDTO dto = FollowCancelCountDTO.builder()
                .userId(body.getUserId())
                .followUserId(body.getFollowUserId())
                .type(FollowCancelTypeEnum.CANCEL.getType())
                .build();
        sendCountMq(dto);
    }


    /**
     * 计数服务异步处理
     * @param dto
     */
    private void sendCountMq(FollowCancelCountDTO dto) {

        String jsonString = JSON.toJSONString(dto);
        MessageBuilder<String> stringMessageBuilder = MessageBuilder.withPayload(jsonString);

        // 关注或者取消计数
        rocketMQTemplate.asyncSend(RocketMqConstant.FOLLOW_CANCEL_COUNT_TOPIC,stringMessageBuilder,new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                logger.info("==>计数服务关注或者取消数MQ发送成功: {}", sendResult);
            }
            @Override
            public void onException(Throwable throwable) {
                logger.info("==>请求计数服务关注或者取消数MQ发送异常: ", throwable);
            }
        });
        // 粉丝计数处理
        rocketMQTemplate.asyncSend(RocketMqConstant.FANS_COUNT_TOPIC,stringMessageBuilder,new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                logger.info("==>计数服务粉丝数MQ发送成功: {}", sendResult);
            }
            @Override
            public void onException(Throwable throwable) {
                logger.info("==>请求计数服务粉丝数MQ发送异常: ", throwable);
            }
        });


    }
}
